package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Stream_Windows {
  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val  conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    val socektDS: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)
    //窗口数据的范围必须为采集周期的整数倍，步长也必须为采集周期的整数
    val word: DStream[(String, Int)] = socektDS.map((_, 1))
    //参数解读：第一个参数是窗口范围。第二个参数是窗口的步长（注意步长一定要是周期的整数倍）
    word.window( Seconds(6),Seconds(6)).print()
    //启动采集器
    ssc.start()
    //Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
